package com.example.demo.handler.imp;

import com.example.demo.base.aeon.PropertyBench;
import com.example.demo.entity.model.SmsReport;
import com.example.demo.handler.Action;
import com.example.demo.util.MiscTools;
import com.example.demo.util.PartitionsTool;
import com.example.demo.util.ScheduledExecutor;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;


import java.util.ArrayList;
import java.util.List;
import java.util.stream.Collectors;

/**
 * Created by wanghuajie on 2018/9/5.
 */
public class CmppInboundHandlerImpl {
    private static final Logger LOGGER = LoggerFactory.getLogger(CmppInboundHandlerImpl.class);

    public void submitReportReach(SmsReport smsReport) {
        LOGGER.info("获取到状态报告");
        REPORT_SUBMITTER.submitAdd(() -> smsReport);
    }

    /**
     * 状态报告控制
     * kafka落地
     */
    private static final ScheduledExecutor<SmsReport> REPORT_SUBMITTER = new ScheduledExecutor<>(
            new Action<SmsReport>() {
                private List<SmsReport> reports = new ArrayList<>();

                @Override
                public void add(SmsReport t) {
                    reports.add(t);
                }

                @Override
                public void consume() {
                    if (reports.size() == 0) {
                        return;
                    }
                    reports.stream().collect(Collectors.groupingBy(c -> PartitionsTool.getDeliverPartition(c.getMsgId()))).
                            forEach((partition, rpts) -> MiscTools.syncSendList2Kafka(
                                    rpts,
                                    (rpt) -> new ProducerRecord<>(MiscTools.AGGREGATE_TOPIC, partition, null, PropertyBench.KAFKA_SPLIT_CHAR + rpt.kafkaStr()),
                                    (okRpt) -> {
                                        try {
                                            //CmppHelper.sendDeliverResp(okRpt.getCtx(), okRpt.getCmppDeliver());
                                            LOGGER.info("发送短信报告响应[{}]");
                                        } catch (Throwable e) {
                                            LOGGER.error("无法发送状态报告确认消息,状态报告MsgId[{}]", okRpt.getMsgId(), e);
                                        }
                                    }

                            ));
                }

                @Override
                public void clear() {
                    reports.clear();
                }
            }
            , "report");


}
